// https://dev.mysql.com/doc/index-other.html
const { ErrorModel, SuccessModel, arrayToObject } = require('../utils/index')
const logger = require('../utils/logger')
const { DataETLModel } = require('../models/data_etl')
const { Query } = require('./../utils/query')
const { etlNodeRunAction } = require('../utils/etl')
const { etlArrayToTree } = require('../utils/tree')

class DataETLController {
  /**
   * @description 添加ETL
   * @param {*} name
   * @param {*} data_etl_id
   * @param {*} remark
   */
  static async dataETLAdd (req, res, next) {
    const { name, remark, data_source_id } = req.body
    const { user_id: op_user_id } = req.user
    logger.info(`DataETLController`, `dataETLAdd`, `req.body`, req.body)
    if (!name || !data_source_id) {
      res.json(new ErrorModel([], `缺少必填项`))
      return
    }
    try {
      const ETLInfo = await DataETLModel.findByName(name)
      if (ETLInfo) {
        res.json(new ErrorModel([], `当前ETL已存在`))
        return
      }
      const result = await DataETLModel.create(
        name,
        data_source_id,
        op_user_id,
        remark
      )
      logger.info(`DataETLController`, `dataETLAdd`, `result`, result)
      res.json(new SuccessModel([], `新增ETL成功`))
    } catch (error) {
      logger.error(`DataETLController`, `dataETLAdd`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }
  /**
   * @description 更新ETL
   * @param {*} data_etl_id
   * @param {*} name
   * @param {*} data_etl_id
   * @param {*} remark
   */
  static async dataETLUpd (req, res, next) {
    const { data_etl_id, name, data_source_id, remark } = req.body
    const { user_id: op_user_id } = req.user
    logger.info(`DataETLController`, `dataETLUpd`, `req.body`, req.body)
    if (!data_etl_id || !name) {
      res.json(new ErrorModel([], `请输入必填项`))
      return
    }
    try {
      const ETLInfo = await DataETLModel.findByNameAndId(name, data_etl_id)
      if (ETLInfo) {
        res.json(new ErrorModel([], `ETL已存在`))
        return
      }
      const result = await DataETLModel.update(
        name,
        data_source_id,
        op_user_id,
        data_etl_id,
        remark
      )
      logger.info(`DataETLController`, `dataETLUpd`, `result`, result)
      res.json(new SuccessModel([], `ETL更新成功`))
    } catch (error) {
      logger.error(`DataETLController`, `dataETLUpd`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }
  /**
   * @description 删除ETL
   * @param {*} data_etl_id
   */
  static async dataETLDel (req, res, next) {
    const { data_etl_id } = req.body
    const { user_id: op_user_id } = req.user
    logger.info(`DataETLController`, `dataETLDel`, `req.body`, req.body)
    if (!data_etl_id) {
      res.json(new ErrorModel([], `data_etl_id 获取失败`))
      return
    }
    try {
      const result = await DataETLModel.delete(op_user_id, data_etl_id)
      logger.info(`DataETLController`, `dataETLDel`, `result`, result)
      res.json(new SuccessModel([], `删除ETL成功`))
    } catch (error) {
      logger.error(`DataETLController`, `dataETLDel`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }
  /**
   * @description 更新ETL
   * @param {*} data_etl_id
   * @param {*} name
   * @param {*} data_etl_id
   * @param {*} remark
   */
  static async dataETLSaveDraft (req, res, next) {
    const { data_etl_id, content } = req.body
    const { user_id: op_user_id } = req.user
    logger.info(`DataETLController`, `dataETLUpd`, `req.body`, req.body)
    if (!data_etl_id || !content) {
      res.json(new ErrorModel([], `请输入必填项`))
      return
    }
    try {
      const result = await DataETLModel.save(content, op_user_id, data_etl_id, 0)
      logger.info(`DataETLController`, `dataETLUpd`, `result`, result)
      res.json(new SuccessModel([], `ETL更新成功`))
    } catch (error) {
      logger.error(`DataETLController`, `dataETLUpd`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }
  /**
   * @description ETL分页列表
   * @param {*} data_etl_id
   */
  static async dataETLList (req, res, next) {
    const { limit, size = 1, name, remark } = req.query
    logger.info(`DataETLController`, `dataETLList`, `req.query`, req.query)
    try {
      const { result, count } = await DataETLModel.list(
        limit,
        size,
        name,
        remark
      )
      logger.info(`DataETLController`, `dataETLList`, `result`, result, count)
      res.json(
        new SuccessModel(
          {
            result,
            total: count[0].count,
            limit: Number(limit),
            size: Number(size),
          },
          `获取ETL数据成功`
        )
      )
    } catch (error) {
      logger.error(`DataETLController`, `dataETLList`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }

  /**
   * @description ETL总列表
   */
  static async dataETLAllList (req, res, next) {
    try {
      const result = await DataETLModel.allList()
      logger.info(`DataETLController`, `dataETLAllList`, `result`, result)
      res.json(new SuccessModel({ result }, `获取ETL数据成功`))
    } catch (error) {
      logger.error(`DataETLController`, `dataETLAllList`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }
  static async saveETLContent (req, res, next) {
    const { content, data_etl_id } = req.body
    const { user_id: op_user_id } = req.user
    logger.info(`DataETLController`, `dataETLAdd`, `req.body`, req.body)
    if (!content || !data_etl_id) {
      res.json(new ErrorModel([], `缺少必填项`))
      return
    }
    try {
      const result = await DataETLModel.updateContent(
        stringContent,
        data_source_id,
        op_user_id
      )
      logger.info(`DataETLController`, `dataETLAdd`, `result`, result)
      res.json(new SuccessModel([], `保存ETL成功`))
    } catch (error) {
      logger.error(`DataETLController`, `dataETLAdd`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }
  /**
   * @description 获取ETL数据
   * @param {*} data_etl_id
   */
  static async dataETLByID (req, res, next) {
    const { data_etl_id } = req.query
    logger.info(`DataETLController`, `dataETLByID`, `req.body`, req.query)
    if (!data_etl_id) {
      res.json(new ErrorModel([], `查询失败,获取不到data_etl_id`))
      return
    }
    try {
      const ETLInfo = await DataETLModel.findById(Number(data_etl_id))
      if (!ETLInfo) {
        res.json(new ErrorModel([], `查询失败`))
        return
      }
      res.json(new SuccessModel(ETLInfo, '查询成功'))
    } catch (error) {
      logger.error(`DataETLController`, `dataETLByID`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }
  static async dataETLMysqlTables (req, res, next) {
    const { data_etl_id } = req.query
    logger.info(`DataETLController`, `dataETLByID`, `req.body`, req.body)
    if (!data_etl_id) {
      res.json(new ErrorModel([], `查询失败,获取不到data_etl_id`))
      return
    }
    try {
      // source info and etl info
      const info = await DataETLModel.findAllInfoById(Number(data_etl_id))
      if (!info) {
        res.json(new ErrorModel([], `查询失败`))
        return
      }
      const { host, port, user, password, db_name, db_type } = info[0]
      const mysqlInfo = {
        host: host,
        port: port,
        user: user,
        password: password,
        database: db_name,
      }
      switch (db_type) {
        case 'MySQL':
          let mycode = `
            SELECT
                TABLE_NAME as label,
                TABLE_NAME as value
            FROM
                information_schema.TABLES
            WHERE
                TABLE_SCHEMA = '${db_name}';`
          Query.getMySQLData(mysqlInfo, mycode)
            .then(async (rows) => {
              logger.info(`DataSQLController`, `dataSQLCheck`, rows)
              res.json(new SuccessModel(rows, `success`))
            })
            .catch(async (err) => {
              logger.error(`DataSQLController`, `dataSQLCheck`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })
          break
        case 'PostgreSQL':
          const pgInfo = {
            host: host, // 数据库所在IP
            user: user, // 数据库用户名
            database: db_name, // 数据库
            password: password, // 数据库密码
            port: port, // 连接端口
          }
          let pcode = `
            SELECT
              tablename as label,
              tablename as value
            FROM
                pg_tables
            WHERE
                tableowner = '${user}'
            AND schemaname = 'public';`
          Query.getPostgreSQLData(pgInfo, pcode)
            .then(async (rows) => {
              logger.info(`DataSQLController`, `dataSQLByID`, rows)
              res.json(new SuccessModel(rows, `success`))
            })
            .catch(async (err) => {
              logger.error(`DataSQLController`, `dataSQLByID`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })

          break
        case 'Oracle':
          let osql = `select table_name as value, table_name as label
          from user_tables
         order by table_name`
          const oracleInfo = {
            user: user, //用户名
            password: password, //密码
            connectString: `${host}:${port}/${db_name}`,
          }
          Query.getOracleData(oracleInfo, osql)
            .then(async (rows) => {
              logger.info(`DataSQLController`, `dataSQLByID`, rows)
              res.json(new SuccessModel(rows, `success`))
            })
            .catch(async (err) => {
              logger.error(`DataSQLController`, `dataSQLByID`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })
          break
        default:
          res.json(new ErrorModel([], '暂不支持此类型数据库'))
          break
      }
    } catch (error) {
      logger.error(`DataETLController`, `dataETLByID`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }
  static async dataETLMysqlFields (req, res, next) {
    const { data_etl_id, table } = req.query
    logger.info(`DataETLController`, `dataETLByID`, `req.body`, req.body)
    if (!data_etl_id) {
      res.json(new ErrorModel([], `查询失败,获取不到data_etl_id`))
      return
    }
    try {
      // source info and etl info
      const info = await DataETLModel.findAllInfoById(Number(data_etl_id))
      if (!info) {
        res.json(new ErrorModel([], `查询失败`))
        return
      }
      const { host, port, user, password, db_name, db_type } = info[0]
      const mysqlInfo = {
        host: host,
        port: port,
        user: user,
        password: password,
        database: db_name,
      }
      switch (db_type) {
        case 'MySQL':
          Query.getMySQLData(mysqlInfo, `
          SELECT
              COLUMN_NAME,
              DATA_TYPE,
              COLUMN_COMMENT,
              COLUMN_NAME as alias
          FROM
              information_schema.COLUMNS
          WHERE
              TABLE_SCHEMA = '${db_name}' AND TABLE_NAME='${table}';`)
            .then(async (rows) => {
              logger.info(`DataSQLController`, `dataSQLCheck`, rows)
              let result = {
                keys: ['COLUMN_NAME', 'DATA_TYPE', 'COLUMN_COMMENT'],
                rows,
              }
              res.json(new SuccessModel(result, `success`))
            })
            .catch(async (err) => {
              logger.error(`DataSQLController`, `dataETLMysqlFields`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })
          break
        case 'PostgreSQL':
          const pgInfo = {
            host: host, // 数据库所在IP
            user: user, // 数据库用户名
            database: db_name, // 数据库
            password: password, // 数据库密码
            port: port, // 连接端口
          }
          Query.getPostgreSQLData(pgInfo, `
            select
              a.attname as "COLUMN_NAME",
              t.typname as "DATA_TYPE",
              col_description(a.attrelid,a.attnum) as "COLUMN_COMMENT",
              a.attname as alias
          from
              pg_class c,
              pg_attribute a,
              pg_type t
          where
              a.attrelid = c.oid
              and a.atttypid = t.oid
              and a.attnum > 0
              and c.relname = '${table}';`)
            .then(async (rows) => {
              logger.info(`DataSQLController`, `dataSQLCheck`, rows)
              let result = {
                keys: ['COLUMN_NAME', 'DATA_TYPE', 'COLUMN_COMMENT'],
                rows,
              }
              res.json(new SuccessModel(result, `success`))
            })
            .catch(async (err) => {
              logger.error(`DataSQLController`, `dataETLMysqlFields`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })

          break
        case 'Oracle':
          const oracleInfo = {
            user: user, //用户名
            password: password, //密码
            connectString: `${host}:${port}/${db_name}`,
          }
          Query.getOracleData(oracleInfo, `
            select utc.Table_Name as "COLUMN_NAME",
            utc.Table_Name as "alias",
            utc.DATA_TYPE as "DATA_TYPE",
            ucc.COMMENTS as "COLUMN_COMMENT"
            from user_tab_columns  utc
            left join  user_col_comments ucc on utc.COLUMN_NAME = ucc.COLUMN_NAME   
            where utc.Table_Name='${table}';`)
            .then(async (rows) => {
              logger.info(`DataSQLController`, `dataSQLCheck`, rows)
              let result = {
                keys: ['COLUMN_NAME', 'DATA_TYPE', 'COLUMN_COMMENT'],
                rows,
              }
              res.json(new SuccessModel(result, `success`))
            })
            .catch(async (err) => {
              logger.error(`DataSQLController`, `dataETLMysqlFields`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })
          break
        default:
          res.json(new ErrorModel([], '暂不支持此类型数据库'))
          break
      }
    } catch (error) {
      logger.error(`DataETLController`, `dataETLByID`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }
  static async dataETLNodeData (req, res, next) {
    const { node, data_etl_id } = req.body
    if (!data_etl_id) {
      res.json(new ErrorModel([], `查询失败,获取不到data_etl_id`))
      return
    }
    if (!node) {
      res.json(new ErrorModel([], `查询失败,获取不到节点数据`))
      return
    }
    try {
      // source info and etl info
      const info = await DataETLModel.findAllInfoById(Number(data_etl_id))
      if (!info) {
        res.json(new ErrorModel([], `查询失败`))
        return
      }
      const { host, port, user, password, db_name, db_type } = info[0]
      const mysqlInfo = {
        host: host,
        port: port,
        user: user,
        password: password,
        database: db_name,
      }
      let sql = etlNodeRunAction(JSON.parse(node), db_type)
      console.log(sql);
      switch (db_type) {
        case 'MySQL':
          Query.getMySQLData(mysqlInfo, sql)
            .then(async (rows) => {
              logger.info(`DataSQLController`, `dataETLNodeData`, rows)
              res.json(new SuccessModel(rows, `success`))
            })
            .catch(async (err) => {
              logger.error(`DataSQLController`, `dataETLNodeData`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })
          break
        case 'PostgreSQL':
          const pgInfo = {
            host: host, // 数据库所在IP
            user: user, // 数据库用户名
            database: db_name, // 数据库
            password: password, // 数据库密码
            port: port, // 连接端口
          }
          Query.getPostgreSQLData(pgInfo, sql)
            .then(async (rows) => {
              logger.info(`DataSQLController`, `dataETLNodeData`, rows)
              res.json(new SuccessModel(rows, `success`))
            })
            .catch(async (err) => {
              logger.error(`DataSQLController`, `dataETLNodeData`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })

          break
        case 'Oracle':
          const oracleInfo = {
            user: user, //用户名
            password: password, //密码
            connectString: `${host}:${port}/${db_name}`,
          }
          Query.getOracleData(oracleInfo, sql)
            .then(async (rows) => {
              logger.info(`DataSQLController`, `dataETLNodeData`, rows)
              res.json(new SuccessModel(rows, `success`))
            })
            .catch(async (err) => {
              logger.error(`DataSQLController`, `dataETLNodeData`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })
          break
        default:
          res.json(new ErrorModel([], '暂不支持此类型数据库'))
          break
      }
    } catch (error) {
      logger.error(`DataETLController`, `dataETLNodeData`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }
  static async dataETLSaveAndRun (req, res, next) {
    const { content, data_etl_id } = req.body
    const { user_id: op_user_id } = req.user
    if (!data_etl_id) {
      res.json(new ErrorModel([], `查询失败,获取不到data_etl_id`))
      return
    }
    if (!content) {
      res.json(new ErrorModel([], `查询失败,获取不到节点数据`))
      return
    }
    try {
      // source info and etl info
      const info = await DataETLModel.findAllInfoById(Number(data_etl_id))
      if (!info) {
        res.json(new ErrorModel([], `查询失败`))
        return
      }
      const { host, port, user, password, db_name, db_type } = info[0]
      const mysqlInfo = {
        host: host,
        port: port,
        user: user,
        password: password,
        database: db_name,
      }
      const { lineList = [], nodeList = [] } = JSON.parse(content)
      let outputNode = nodeList.find(item => item.type === 'output');
      let nodeTree = etlArrayToTree(
        lineList,
        outputNode.id,
        "from",
        "to",
        nodeList
      )
      outputNode.children = nodeTree
      let columns = outputNode.config.fields.map(item => item.alias)
      let sql = etlNodeRunAction(outputNode, db_type)
      switch (db_type) {
        case 'MySQL':
          Query.getMySQLData(mysqlInfo, sql)
            .then(async (rows) => {
              logger.info(`DataSQLController`, `dataETLNodeData`, rows)
              try {
                const result = await DataETLModel.save(content, op_user_id, data_etl_id, 1)
                logger.info(`DataETLController`, `dataETLSaveAndRun`, `result`, result)
                res.json(new SuccessModel({ columns, rows }, `ETL保存成功`))
              } catch (error) {
                logger.error(`DataETLController`, `dataETLSaveAndRun`, `error`, error)
                res.json(new ErrorModel([], error.message))
              }
            })
            .catch(async (err) => {
              logger.error(`DataSQLController`, `dataETLNodeData`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })
          break
        case 'PostgreSQL':
          const pgInfo = {
            host: host, // 数据库所在IP
            user: user, // 数据库用户名
            database: db_name, // 数据库
            password: password, // 数据库密码
            port: port, // 连接端口
          }
          Query.getPostgreSQLData(pgInfo, sql)
            .then(async (rows) => {
              try {
                const result = await DataETLModel.save(content, op_user_id, data_etl_id, 1)
                logger.info(`DataETLController`, `dataETLSaveAndRun`, `result`, result)
                res.json(new SuccessModel({ columns, rows }, `ETL保存成功`))
              } catch (error) {
                logger.error(`DataETLController`, `dataETLSaveAndRun`, `error`, error)
                res.json(new ErrorModel([], error.message))
              }
            })
            .catch(async (err) => {
              logger.error(`DataSQLController`, `dataETLNodeData`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })

          break
        case 'Oracle':
          const oracleInfo = {
            user: user, //用户名
            password: password, //密码
            connectString: `${host}:${port}/${db_name}`,
          }
          Query.getOracleData(oracleInfo, sql)
            .then(async (rows) => {
              try {
                const result = await DataETLModel.save(content, op_user_id, data_etl_id, 1)
                logger.info(`DataETLController`, `dataETLSaveAndRun`, `result`, result)
                res.json(new SuccessModel({ columns, rows }, `ETL保存成功`))
              } catch (error) {
                logger.error(`DataETLController`, `dataETLSaveAndRun`, `error`, error)
                res.json(new ErrorModel([], error.message))
              }
            })
            .catch(async (err) => {
              logger.error(`DataSQLController`, `dataETLNodeData`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })
          break
        default:
          res.json(new ErrorModel([], '暂不支持此类型数据库'))
          break
      }
    } catch (error) {
      logger.error(`DataETLController`, `dataETLNodeData`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }
  static async dataETLTableData (req, res, next) {
    const { table, limit = 0, data_etl_id, fields } = req.body
    let sqlField = fields.map((item) => {
      return `${table}.${item.COLUMN_NAME} as ${item.alias}`
    })
    if (!data_etl_id) {
      res.json(new ErrorModel([], `查询失败,获取不到data_etl_id`))
      return
    }
    if (!table) {
      res.json(new ErrorModel([], `查询失败,获取不到表名`))
      return
    }
    try {
      // source info and etl info
      const info = await DataETLModel.findAllInfoById(Number(data_etl_id))
      if (!info) {
        res.json(new ErrorModel([], `查询失败`))
        return
      }
      const { host, port, user, password, db_name, db_type } = info[0]
      switch (db_type) {
        case 'MySQL':
          let limitMySql = ''
          if (limit) {
            limitMySql = `LIMIT 0,${limit}`
          }
          const mysqlInfo = {
            host: host,
            port: port,
            user: user,
            password: password,
            database: db_name,
          }
          let mycode = `SELECT ${sqlField.join(',')} FROM ${table} ${limitMySql};`
          Query.getMySQLData(
            mysqlInfo,
            mycode
          ).then(async (rows) => {
            logger.info(`dataETLTableData`, `dataSQLCheck`, rows)
            res.json(new SuccessModel(rows, `success`))
          })
            .catch(async (err) => {
              logger.error(`dataETLTableData`, `dataSQLCheck`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })
          break;
        case "PostgreSQL":
          let limitPSql = ''
          if (limit) {
            limitPSql = `LIMIT ${limit} offset 0`
          }
          const pgInfo = {
            host: host, // 数据库所在IP
            user: user, // 数据库用户名
            database: db_name, // 数据库
            password: password, // 数据库密码
            port: port, // 连接端口
          }
          let pcode = `SELECT ${sqlField.join(',')} FROM ${table} ${limitPSql};`
          Query.getPostgreSQLData(pgInfo, pcode)
            .then(async (rows) => {
              logger.info(`dataETLTableData`, `dataSQLCheck`, rows)
              res.json(new SuccessModel(rows, `success`))
            })
            .catch(async (err) => {
              logger.error(`dataETLTableData`, `dataSQLCheck`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })

          break
        case 'Oracle':
          let limitOSQL = ''
          if (limit) {
            limitOSQL = `rownum <= ${limit}`
          }
          const oracleInfo = {
            user: user, //用户名
            password: password, //密码
            connectString: `${host}:${port}/${db_name}`,
          }
          let ocode = `SELECT ${sqlField.join(',')} FROM ${table} ${limitOSQL};`
          Query.getOracleData(oracleInfo, ocode)
            .then(async (rows) => {
              logger.info(`dataETLTableData`, `dataSQLCheck`, rows)
              res.json(new SuccessModel(rows, `success`))
            })
            .catch(async (err) => {
              logger.error(`dataETLTableData`, `dataSQLCheck`, `error`, err)
              res.json(new ErrorModel([], err.message))
            })
          break
        default:
          res.json(new ErrorModel([], '暂不支持此类型数据库'))
          break;
      }

    } catch (error) {
      logger.error(`DataETLController`, `dataETLByID`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }
  static async dataETLExec (req, res, next) {
    const { data_etl_id } = req.body
    logger.info(`DataETLController`, `dataETLExec`, `req.query`, req.query)
    if (!data_etl_id) {
      res.json(new ErrorModel([], `查询失败,获取不到data_etl_id`))
      return
    }
    try {
      // source info and etl info
      const info = await DataETLModel.findAllInfoById(Number(data_etl_id))
      if (!info) {
        res.json(new ErrorModel([], `查询失败`))
        return
      }
      const { host, port, user, password, db_name, content, status, db_type } = info[0]
      if (status !== 1) {
        res.json(new ErrorModel([], `暂未配置完成！`))
        return
      }
      if (!content) {
        res.json(new ErrorModel([], `查询失败,获取不到节点数据`))
        return
      }
      try {
        // source info and etl info
        const info = await DataETLModel.findAllInfoById(Number(data_etl_id))
        if (!info) {
          res.json(new ErrorModel([], `查询失败`))
          return
        }
        const { host, port, user, password, db_name, db_type } = info[0]
        const { lineList = [], nodeList = [] } = JSON.parse(content)
        let outputNode = nodeList.find(item => item.type === 'output');
        let nodeTree = etlArrayToTree(
          lineList,
          outputNode.id,
          "from",
          "to",
          nodeList
        )
        outputNode.children = nodeTree
        let sql = etlNodeRunAction(outputNode, db_type)
        // const { host, port, user, password, db_name, db_type } = info[0]
        switch (db_type) {
          case 'MySQL':
            const mysqlInfo = {
              host: host,
              port: port,
              user: user,
              password: password,
              database: db_name,
            }
            Query.getMySQLData(mysqlInfo, sql)
              .then(async (rows) => {
                logger.info(`DataSQLController`, `dataETLExec`, rows)
                res.json(new SuccessModel(rows, `success`))
              })
              .catch(async (err) => {
                logger.error(`DataSQLController`, `dataETLExec`, `error`, err)
                res.json(new ErrorModel([], err.message))
              })
            break
          case 'PostgreSQL':
            const pgInfo = {
              host: host, // 数据库所在IP
              user: user, // 数据库用户名
              database: db_name, // 数据库
              password: password, // 数据库密码
              port: port, // 连接端口
            }
            Query.getPostgreSQLData(pgInfo, sql)
              .then(async (rows) => {
                logger.info(`DataSQLController`, `dataETLExec`, rows)
                res.json(new SuccessModel(rows, `success`))
              })
              .catch(async (err) => {
                logger.error(`DataSQLController`, `dataETLExec`, `error`, err)
                res.json(new ErrorModel([], err.message))
              })

            break
          case 'Oracle':
            const oracleInfo = {
              user: user, //用户名
              password: password, //密码
              connectString: `${host}:${port}/${db_name}`,
            }
            Query.getOracleData(oracleInfo, sql)
              .then(async (rows) => {
                logger.info(`DataSQLController`, `dataETLExec`, rows)
                res.json(new SuccessModel(rows, `success`))
              })
              .catch(async (err) => {
                logger.error(`DataSQLController`, `dataETLExec`, `error`, err)
                res.json(new ErrorModel([], err.message))
              })
            break
          default:
            res.json(new ErrorModel([], '暂不支持此类型数据库'))
            break
        }
      } catch (error) {
        logger.error(`DataETLController`, `dataETLExec`, `error`, error)
        res.json(new ErrorModel([], error.message))
      }
    } catch (error) {
      logger.error(`DataETLController`, `dataETLExec`, `error`, error)
      res.json(new ErrorModel([], error.message))
    }
  }
}

module.exports = {
  DataETLController,
}
